package com.wiwf.model.mqtt.config;

import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;

/**
 * @author kehui
 */
@Configuration(proxyBeanMethods = false)
public class BackMqttClientCfg {

	public static final String TOP = "clent";

    @Value("${base.mqtt."+TOP+".url}")
    private String url;
    @Value("${base.mqtt."+TOP+".username}")
    private String username;
    @Value("${base.mqtt."+TOP+".password}")
    private String password;

    @Value("${base.mqtt."+TOP+".receiver.client}")
    private String clientcId;
    @Value("${base.mqtt."+TOP+".receiver.topic}")
    private String defaultcTopic;



    @Bean(TOP+"MqttConnectOptions")
    public MqttConnectOptions getSenderMqttConnectOptions() {
        MqttConnectOptions options = new MqttConnectOptions();

        options.setUserName(username);
        options.setPassword(password.toCharArray());
        options.setServerURIs(new String[]{url});
        options.setCleanSession(true);
        options.setAutomaticReconnect(true);
        options.setConnectionTimeout(100);
        options.setKeepAliveInterval(30);
        return options;
    }
    @Bean(TOP+"SenderMqttClientFactory")
    public MqttPahoClientFactory senderMqttClientFactory(@Autowired @Qualifier(TOP+"MqttConnectOptions") MqttConnectOptions options) {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        factory.setConnectionOptions(options);
        return factory;
    }

    /**
     * MQTT信息通道（消费者）
     */
    @Bean(TOP+"MqttInboundChannel")
    public MessageChannel mqttInboundChannel() {
        DirectChannel channel = new DirectChannel();
        return channel;
    }


    /**
     * MQTT消息订阅绑定（消费者）
     */
    @Bean
    public MessageProducer inbound(
            @Autowired @Qualifier(TOP+"SenderMqttClientFactory") MqttPahoClientFactory mqttPahoClientFactory,
            @Autowired @Qualifier(TOP+"MqttInboundChannel") MessageChannel messageChannel
    ) {
        // 可以同时消费（订阅）多个Topic
        MqttPahoMessageDrivenChannelAdapter adapter =
                new MqttPahoMessageDrivenChannelAdapter( clientcId, mqttPahoClientFactory, defaultcTopic);
        adapter.setCompletionTimeout(5000);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(1);
        // 设置订阅通道
        adapter.setOutputChannel(messageChannel);
        return adapter;
    }

    /**
     * MQTT消息处理器（消费者）
     */
    @Bean
    @ServiceActivator(inputChannel = TOP+"MqttInboundChannel")
    public MessageHandler handler( @Autowired @Qualifier("backMqttReceiverImpl") MessageHandler gdtMqttReceiverImpl) {
        return gdtMqttReceiverImpl;
    }


}
